package com.atguigu.tingshu.search.service.impl;

import co.elastic.clients.elasticsearch.ElasticsearchClient;
import co.elastic.clients.elasticsearch._types.FieldValue;
import co.elastic.clients.elasticsearch._types.SortOrder;
import co.elastic.clients.elasticsearch._types.aggregations.Aggregate;
import co.elastic.clients.elasticsearch._types.query_dsl.TermsQueryField;
import co.elastic.clients.elasticsearch.core.SearchRequest;
import co.elastic.clients.elasticsearch.core.SearchResponse;
import co.elastic.clients.elasticsearch.core.search.CompletionSuggest;
import co.elastic.clients.elasticsearch.core.search.Hit;
import co.elastic.clients.elasticsearch.core.search.Suggestion;
import com.alibaba.fastjson.JSON;
import com.atguigu.tingshu.album.client.AlbumInfoFeignClient;
import com.atguigu.tingshu.album.client.CategoryFeignClient;
import com.atguigu.tingshu.common.constant.RedisConstant;
import com.atguigu.tingshu.common.result.Result;
import com.atguigu.tingshu.common.util.PinYinUtils;
import com.atguigu.tingshu.model.album.*;
import com.atguigu.tingshu.model.search.AlbumInfoIndex;
import com.atguigu.tingshu.model.search.AttributeValueIndex;
import com.atguigu.tingshu.model.search.SuggestIndex;
import com.atguigu.tingshu.query.search.AlbumIndexQuery;
import com.atguigu.tingshu.search.repository.AlbumIndexRepository;
import com.atguigu.tingshu.search.repository.SuggestIndexRepository;
import com.atguigu.tingshu.search.service.SearchService;
import com.atguigu.tingshu.user.client.UserInfoFeignClient;
import com.atguigu.tingshu.vo.album.AlbumInfoVo;
import com.atguigu.tingshu.vo.search.AlbumInfoIndexVo;
import com.atguigu.tingshu.vo.search.AlbumSearchResponseVo;
import com.atguigu.tingshu.vo.user.UserInfoVo;
import jakarta.json.JsonObject;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.redisson.api.RBloomFilter;
import org.redisson.api.RedissonClient;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.elasticsearch.core.suggest.Completion;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Service;
import org.springframework.util.Assert;
import org.springframework.util.CollectionUtils;

import java.io.IOException;
import java.util.*;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.stream.Collectors;


@Slf4j
@Service
@SuppressWarnings({"unchecked", "rawtypes"})
public class SearchServiceImpl implements SearchService {


    @Autowired
    private AlbumInfoFeignClient albumInfoFeignClient;
    @Autowired
    private CategoryFeignClient categoryFeignClient;
    @Autowired
    private UserInfoFeignClient userInfoFeignClient;
    @Autowired
    private AlbumIndexRepository albumIndexRepository;
    @Autowired
    private ThreadPoolExecutor threadPoolExecutor;
    @Autowired
    private ElasticsearchClient elasticsearchClient;
    @Autowired
    private SuggestIndexRepository suggestIndexRepository;
    @Autowired
    private RedisTemplate redisTemplate;
    @Autowired
    private RedissonClient redissonClient;


    @Override
    public void upperAlbum(Long albumId) {

        AlbumInfoIndex albumInfoIndex = new AlbumInfoIndex();

        CompletableFuture<AlbumInfo> albumInfoCompletableFuture = CompletableFuture.supplyAsync(() -> {

            Result<AlbumInfo> albumInfoResult = albumInfoFeignClient.getAlbumInfo(albumId);

            Assert.notNull(albumInfoResult, "返回专辑结果为空");

            AlbumInfo albumInfo = albumInfoResult.getData();

            Assert.notNull(albumInfo, "返回专辑为空");

            BeanUtils.copyProperties(albumInfo, albumInfoIndex);

            return albumInfo;
        },threadPoolExecutor);


        CompletableFuture<Void> categoryCompletableFuture = albumInfoCompletableFuture.thenAcceptAsync(albumInfo -> {
            //远程调用三级分类对象
            Result<BaseCategoryView> categoryViewResult = this.categoryFeignClient.getCategoryView(albumInfo.getCategory3Id());

            Assert.notNull(categoryViewResult, "返回分类结果集为空");

            BaseCategoryView categoryView = categoryViewResult.getData();

            Assert.notNull(categoryView, "返回分类为空");

            albumInfoIndex.setCategory1Id(categoryView.getCategory1Id());
            albumInfoIndex.setCategory2Id(categoryView.getCategory2Id()); //三级分类id之前已经属性拷贝了
        },threadPoolExecutor);


        //更新统计量与得分，默认随机，方便测试
        int num1 = new Random().nextInt(1000);
        int num2 = new Random().nextInt(100);
        int num3 = new Random().nextInt(50);
        int num4 = new Random().nextInt(300);
        albumInfoIndex.setPlayStatNum(num1);
        albumInfoIndex.setSubscribeStatNum(num2);
        albumInfoIndex.setBuyStatNum(num3);
        albumInfoIndex.setCommentStatNum(num4);
        double hotScore = num1*0.2 + num2*0.3 + num3*0.4 + num4*0.1;
        //  设置热度排名
        albumInfoIndex.setHotScore(hotScore);


        CompletableFuture<Void> attrCompletableFuture = CompletableFuture.runAsync(() -> {
            //赋值属性数据 属性数据类型为Nested
            Result<List<AlbumAttributeValue>> albumAttributeValueResult = albumInfoFeignClient.findAlbumAttributeValue(albumId);
            Assert.notNull(albumAttributeValueResult, "专辑属性结果集为空");
            List<AlbumAttributeValue> albumAttributeValueList = albumAttributeValueResult.getData();
            Assert.notNull(albumAttributeValueList, "专辑属性集合为空");
            List<AttributeValueIndex> attributeValueIndexList = albumAttributeValueList.stream().map(item -> {
                AttributeValueIndex attributeValueIndex = new AttributeValueIndex();
                BeanUtils.copyProperties(item, attributeValueIndex);
                return attributeValueIndex;
            }).collect(Collectors.toList());
            albumInfoIndex.setAttributeValueIndexList(attributeValueIndexList);
        },threadPoolExecutor);


        CompletableFuture<Void> userCompletableFuture = albumInfoCompletableFuture.thenAcceptAsync(albumInfo -> {
            //主播
            Result<UserInfoVo> userInfoVoResult = this.userInfoFeignClient.getUserInfoVo(albumInfo.getUserId());
            Assert.notNull(userInfoVoResult, "用户结果集为空");
            UserInfoVo userInfoVo = userInfoVoResult.getData();
            Assert.notNull(userInfoVo, "用户对象为空");
            albumInfoIndex.setAnnouncerName(userInfoVo.getNickname());
        },threadPoolExecutor);

        CompletableFuture.allOf(albumInfoCompletableFuture,categoryCompletableFuture,attrCompletableFuture,userCompletableFuture).join();

        albumIndexRepository.save(albumInfoIndex);


        //  上架添加提词数据.
        //  创建对象 专辑标题提词
        SuggestIndex suggestIndex = new SuggestIndex();
        suggestIndex.setId(UUID.randomUUID().toString().replaceAll("-",""));
        suggestIndex.setTitle(albumInfoIndex.getAlbumTitle());
        suggestIndex.setKeyword(new Completion(new String[]{albumInfoIndex.getAlbumTitle()}));
        suggestIndex.setKeywordPinyin(new Completion(new String[]{PinYinUtils.toHanyuPinyin(albumInfoIndex.getAlbumTitle())}));
        suggestIndex.setKeywordSequence(new Completion(new String[]{PinYinUtils.getFirstLetter(albumInfoIndex.getAlbumTitle())}));
        this.suggestIndexRepository.save(suggestIndex);

        //  专辑简介提词
        SuggestIndex albumIntroSuggestIndex = new SuggestIndex();
        albumIntroSuggestIndex.setId(UUID.randomUUID().toString().replaceAll("-",""));
        albumIntroSuggestIndex.setTitle(albumInfoIndex.getAlbumIntro());
        albumIntroSuggestIndex.setKeyword(new Completion(new String[]{albumInfoIndex.getAlbumIntro()}));
        albumIntroSuggestIndex.setKeywordPinyin(new Completion(new String[]{PinYinUtils.toHanyuPinyin(albumInfoIndex.getAlbumIntro())}));
        albumIntroSuggestIndex.setKeywordSequence(new Completion(new String[]{PinYinUtils.getFirstLetter(albumInfoIndex.getAlbumIntro())}));
        this.suggestIndexRepository.save(albumIntroSuggestIndex);

        //	将新增的商品SKUID存入布隆过滤器
        //	获取布隆过滤器，将新增skuID存入布隆过滤器
        RBloomFilter<Object> bloomFilter = redissonClient.getBloomFilter(RedisConstant.ALBUM_BLOOM_FILTER);
        bloomFilter.add(albumId);

    }

    @Override
    public void lowerAlbum(Long albumId) {

        albumIndexRepository.deleteById(albumId);

    }


    @Override
    public AlbumSearchResponseVo search(AlbumIndexQuery albumIndexQuery) {

        // 获取searchRequest对象
        SearchRequest searchRequest = this.queryBuildDsl(albumIndexQuery);

        SearchResponse<AlbumInfoIndex> searchResponse = null;
        try {
            searchResponse = elasticsearchClient.search(searchRequest, AlbumInfoIndex.class);
        } catch (IOException e) {
            throw new RuntimeException(e);
        }

        //获取返回结果集
        AlbumSearchResponseVo albumSearchResponseVo = this.parseResultData(searchResponse);

        //默认第一页
        albumSearchResponseVo.setPageNo(albumIndexQuery.getPageNo());
        //每页显示10条
        albumSearchResponseVo.setPageSize(albumIndexQuery.getPageSize());

        //总页数
        long totalPages = (albumSearchResponseVo.getTotal() + albumIndexQuery.getPageSize() - 1) / albumIndexQuery.getPageSize();
        albumSearchResponseVo.setTotalPages(totalPages);

        return albumSearchResponseVo;
    }

    /**
     * 获取返回结果集
     * @param searchResponse
     * @return
     */
    private AlbumSearchResponseVo parseResultData(SearchResponse<AlbumInfoIndex> searchResponse) {
        List<Hit<AlbumInfoIndex>> hitList = searchResponse.hits().hits();

        List<AlbumInfoIndexVo> albumInfoIndexList = hitList.stream().map(item -> {

            AlbumInfoIndexVo vo = new AlbumInfoIndexVo();

            AlbumInfoIndex albumInfoIndex = item.source();

            BeanUtils.copyProperties(albumInfoIndex, vo);

            //判断是否是关键字查询 如果是就高亮
            if (null != item.highlight().get("albumTitle")){
                //  获取高亮数据
                String albumTitle = item.highlight().get("albumTitle").get(0);
                //  赋值高亮数据
                vo.setAlbumTitle(albumTitle);
            }

            return vo;
        }).collect(Collectors.toList());

        AlbumSearchResponseVo searchResponseVo = new AlbumSearchResponseVo();

        searchResponseVo.setList(albumInfoIndexList);

        //设置总记录数
        searchResponseVo.setTotal(searchResponse.hits().total().value());

        return searchResponseVo;
    }


    /**
     * 获取searchRequest对象
     * @param albumIndexQuery
     * @return
     */
    private SearchRequest queryBuildDsl(AlbumIndexQuery albumIndexQuery) {

        SearchRequest.Builder searchRequestBuilder = new SearchRequest.Builder();

        //判断当前用户是通过哪个字段进行检索的
        if (StringUtils.isNoneBlank(albumIndexQuery.getKeyword())) {

            //query -- bool -- should -- match
            searchRequestBuilder.query(q -> q.bool(b -> b.should(s -> s.match(m -> m.field("albumTitle")
                            .query(albumIndexQuery.getKeyword())))
                    .should(s -> s.match(m -> m.field("albumIntro")
                            .query(albumIndexQuery.getKeyword())))));
            //设置高亮
            searchRequestBuilder.highlight(h -> h.fields("albumTitle",
                    f->f.preTags("<span style=color:red>").postTags("</span>")));

        }

        //判断用户是否根据用户Id检索
        if (albumIndexQuery.getCategory1Id() != null) {

            searchRequestBuilder.query(q -> q.bool(b -> b.filter(f -> f.term(t -> t.field("category1Id").value(albumIndexQuery.getCategory1Id())))));
        }
        if (albumIndexQuery.getCategory2Id() != null) {

            searchRequestBuilder.query(q -> q.bool(b -> b.filter(f -> f.term(t -> t.field("category2Id").value(albumIndexQuery.getCategory2Id())))));
        }
        if (albumIndexQuery.getCategory3Id() != null) {

            searchRequestBuilder.query(q -> q.bool(b -> b.filter(f -> f.term(t -> t.field("category3Id").value(albumIndexQuery.getCategory3Id())))));
        }

        //属性数据类型为nested(能够允许数组独立的检索查询)
        List<String> attributeList = albumIndexQuery.getAttributeList();

        if (!CollectionUtils.isEmpty(attributeList)) {

            for (String attribute : attributeList) {

                //属性数据格式为 属性id:属性值Id
                String[] split = attribute.split(":");

                if (split != null && split.length == 2) {
                    //query--nested--path--query--bool--filter-term  但条件只通过属性和属性值查询
//                    searchRequestBuilder.query(q -> q.nested(n -> n.path("attributeValueIndexList")
//                            .query(q1 -> q1.bool(b -> b.filter(f -> f.term(t -> t.field("attributeValueIndexList.attributeId").value(split[0])))
//                                    .filter(f -> f.term(t -> t.field("attributeValueIndexList.valueId").value(split[1])))
//                            ))));

                    //  多条件查询！
                    searchRequestBuilder.query(q -> q.bool(b -> b.filter(f -> f.nested(n -> n.path("attributeValueIndexList")
                            .query(qy -> qy.bool(bl -> bl.filter(fl -> fl.term(t -> t.field("attributeValueIndexList.attributeId").value(split[0])))
                                    .filter(fl -> fl.term(t -> t.field("attributeValueIndexList.valueId").value(split[1])))
                            ))
                    ))));
                }
            }
        }

        //排序（综合排序[1:desc] 播放量[2:desc] 发布时间[3:desc]；asc:升序 desc:降序）
        String order = albumIndexQuery.getOrder();  // order的数据格式为：order=1:desc 或者是 2:desc
        if (StringUtils.isNotBlank(order)) {
            String[] split = order.split(":");

            if (split.length == 2 && null != split){

                String orderFiled = null;

                switch (split[0]) {
                    case "1":
                        orderFiled = "hotScore";
                        break;
                    case "2":
                        orderFiled = "playStatNum";
                        break;
                    case "3":
                        orderFiled = "createTime";
                        break;
                }

                //设置排序字段和规则 构建DSL语句
                String finalOrderFiled = orderFiled;
                searchRequestBuilder.sort(s -> s.field(f ->f.field(finalOrderFiled).order(split[1].equals("asc") ? SortOrder.Asc : SortOrder.Desc)));
            }
        }

        //设置分页
        Integer from = (albumIndexQuery.getPageNo() - 1) * albumIndexQuery.getPageSize();
        searchRequestBuilder.from(from);
        searchRequestBuilder.size(albumIndexQuery.getPageSize());

        //主要是封装DSL语句
        SearchRequest searchRequest = searchRequestBuilder.build();

        System.out.println("DSL:\t" + searchRequest.toString());

        return searchRequest;

    }


    @Override
    public List<Map<String, Object>> channel(Long category1Id) {

        //先根据一级分类Id找到三级分类对象数据
        Result<List<BaseCategory3>> baseCategory3ListResult = categoryFeignClient.findTopBaseCategory3(category1Id);
        Assert.notNull(baseCategory3ListResult,"三级分类结果集为空");
        List<BaseCategory3> baseCategory3List = baseCategory3ListResult.getData();
        Assert.notNull(baseCategory3List,"三级分类集合为空");

        //将baseCategory3List转成map key=category3Id  value=BaseCategory3对象
        Map<Long, BaseCategory3> category3IdToMap = baseCategory3List.stream().collect(Collectors.toMap(BaseCategory3::getId, baseCategory3 -> baseCategory3));

        List<Long> category3IdList = baseCategory3List.stream().map(id -> id.getId()).collect(Collectors.toList());

        //  将这个idList进行转换
        List<FieldValue> valueList = category3IdList.stream().map(item -> FieldValue.of(item)).collect(Collectors.toList());

        //  调用查询方法: 法一：
//        SearchRequest.Builder request = new SearchRequest.Builder();
//        request.index("albuminfo").query(q->q.terms(f->f.field("category3Id").terms(new TermsQueryField.Builder().value(valueList).build())));
//        request.aggregations("groupByCategory3IdAgg",a->a.terms(t->t.field("category3Id"))
//                .aggregations("topTenHotScoreAgg",a1->a1.topHits(s->s.size(6).sort(sort->sort.field(f->f.field("hotScore").order(SortOrder.Desc))))));
//
//        //  获取到查询结果集
//        SearchResponse<AlbumInfoIndex> searchResponse = null;
//        try {
//            searchResponse = elasticsearchClient.search(request.build(), AlbumInfoIndex.class);
//        } catch (IOException e) {
//            throw new RuntimeException(e);
//        }

        //法二：
        SearchResponse<AlbumInfoIndex> searchResponse = null;
        try {
            searchResponse = elasticsearchClient.search(f -> f.index("albuminfo").query(q -> q.terms(t -> t.field("category3Id").terms(ts -> ts.value(valueList))))
                            .aggregations("groupByCategory3IdAgg", a -> a.terms(t -> t.field("category3Id"))
                                    .aggregations("topSixHotScoreAgg", ag -> ag.topHits(ft -> ft.size(6)
                                            .sort(s -> s.field(fn -> fn.field("buyStatNum").order(SortOrder.Desc)))))
                            )
                    ,
                    AlbumInfoIndex.class);
        } catch (IOException e) {
            throw new RuntimeException(e);
        }

        //  获取到结果集。
        Map<String, Aggregate> aggregations = searchResponse.aggregations();
        Aggregate groupByCategory3IdAgg = aggregations.get("groupByCategory3IdAgg");
        List<Map<String, Object>> mapList = groupByCategory3IdAgg.lterms().buckets().array().stream().map(bucket -> {
            //  声明一个集合对象
            Map<String, Object> map = new HashMap<>();
            //  获取三级分类Id
            long category3Id = bucket.key();
            //  获取三级分类Id下的专辑数据集合
            List<AlbumInfoIndex> albumInfoIndexList = bucket.aggregations().get("topSixHotScoreAgg").topHits().hits().hits().stream().map(hit -> {
                String strJson = hit.source().toString();
                AlbumInfoIndex albumInfoIndex = JSON.parseObject(strJson, AlbumInfoIndex.class);
                return albumInfoIndex;
            }).collect(Collectors.toList());
            //  存储三级分类对象
            map.put("baseCategory3", category3IdToMap.get(category3Id));
            //  存储专辑集合
            map.put("list", albumInfoIndexList);
            return map;
        }).collect(Collectors.toList());

        //  返回数据
        return mapList;
    }
    @Override
    public List<String> completeSuggest(String keyword) {

        //  查询数据：
        SearchResponse<SuggestIndex> response = null;
        try {
            response = elasticsearchClient.search(f -> f.index("suggestinfo")
                            .suggest(s -> s.suggesters("suggestionKeyword", s1 -> s1.prefix(keyword)
                                                    .completion(c -> c.field("keyword").skipDuplicates(true)
                                                            .size(10).fuzzy(z -> z.fuzziness("auto")))
                                            )
                                            .suggesters("suggestionkeywordPinyin", s2 -> s2.prefix(keyword)
                                                    .completion(c -> c.field("keywordPinyin").skipDuplicates(true)
                                                            .size(10).fuzzy(z -> z.fuzziness("auto")))
                                            )
                                            .suggesters("suggestionkeywordSequence", s2 -> s2.prefix(keyword)
                                                    .completion(c -> c.field("keywordSequence").skipDuplicates(true)
                                                            .size(10).fuzzy(z -> z.fuzziness("auto")))
                                            )
                            )
                    ,
                    SuggestIndex.class);
        } catch (IOException e) {
            throw new RuntimeException(e);
        }

        //  获取数据：
        List<String> keywordList = new ArrayList<>();
        keywordList.addAll(this.getSuggestResult(response, "suggestionKeyword"));
        keywordList.addAll(this.getSuggestResult(response, "suggestionkeywordPinyin"));
        keywordList.addAll(this.getSuggestResult(response, "suggestionkeywordSequence"));

        //  判断这个集合的长度，如果不够10条数据， 如何获取集合的长度
        if (keywordList.size() < 10) {
            //  执行dsl 语句.
            SearchResponse<SuggestIndex> searchResponse = null;
            try {
                searchResponse = elasticsearchClient.search(s -> s.index("suggestinfo").query(q -> q.match(m -> m.field("title").query(keyword))), SuggestIndex.class);
            } catch (IOException e) {
                throw new RuntimeException(e);
            }
            //  List<String> list = searchResponse.hits().hits().stream().map(hit -> hit.source().getTitle()).collect(Collectors.toList());
            for (Hit<SuggestIndex> hit : searchResponse.hits().hits()) {
                String title = hit.source().getTitle();
                keywordList.add(title);
                if (keywordList.size() >= 10) {
                    break;
                }
            }
        }

        return keywordList;
    }

    /**
     * 处理聚合结果集
     * @param response
     * @param keyword
     * @return
     */
    private List<String> getSuggestResult(SearchResponse<SuggestIndex> response, String keyword) {
        //  声明一个集合
        List<String> list = new ArrayList<>();
        //  根据key 获取数据.
        List<Suggestion<SuggestIndex>> suggestionList = response.suggest().get(keyword);
        //  判断
        for (Suggestion<SuggestIndex> suggestIndexSuggestion : suggestionList) {
            list = suggestIndexSuggestion.completion().options().stream().map(suggestIndexCompletionOption -> {
                //  获取title数据
                return suggestIndexCompletionOption.source().getTitle();
            }).collect(Collectors.toList());
        }
        //  返回数据
        return list;
    }


    @Override
    public void updateLatelyAlbumRanking() {
        //  先获取到所有一级分类数据
        Result<List<BaseCategory1>> baseCategory1Result = categoryFeignClient.findAllCategory1();
        Assert.notNull(baseCategory1Result, "一级分类集合结果为空");
        List<BaseCategory1> category1List = baseCategory1Result.getData();
        Assert.notNull(category1List, "一级分类集合为空");

        String[] rankingDimensionArray = new String[]{"hotScore", "playStatNum", "subscribeStatNum", "buyStatNum", "commentStatNum"};
        for (BaseCategory1 baseCategory1 : category1List) {
            for (String ranging : rankingDimensionArray) {
                //  执行dsl语句 获取到结果集.
                SearchResponse<AlbumInfoIndex> response = null;
                try {
                    response = elasticsearchClient.search(f -> f.index("albuminfo")
                            .query(q -> q.term(t -> t.field("category1Id").value(baseCategory1.getId())))
                            .sort(s -> s.field(d -> d.field(ranging).order(SortOrder.Desc)))
                            .size(10), AlbumInfoIndex.class);
                } catch (IOException e) {
                    throw new RuntimeException(e);
                }
                //  获取到当前执行结果集，将数据存储到缓存：
                List<AlbumInfoIndex> albumInfoIndexList = response.hits().hits().stream().map(Hit::source).collect(Collectors.toList());

                //  使用hash 数据结构 hset key field value;
                String rangKey = RedisConstant.RANKING_KEY_PREFIX + baseCategory1.getId();
                this.redisTemplate.boundHashOps(rangKey).put(ranging, albumInfoIndexList);
            }
        }
    }


    @Override
    public List<AlbumInfoIndex> findRankingList(Long category1Id, String dimension) {
        return (List<AlbumInfoIndex>) this.redisTemplate.opsForHash().get(RedisConstant.RANKING_KEY_PREFIX + category1Id, dimension);
    }
}

